package cn.classg.rabbitmq.callback;

import cn.classg.rabbitmq.mq.CallbackInterface;
import cn.classg.rabbitmq.util.JdkSerialization;
import cn.classg.rabbitmq.util.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.task.AsyncTaskExecutor;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 回调
 * @author CLASSG
 */
public class RabbitTemplateCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    public static final Map<Class<?>, CallbackInterface> LAZY_MESSAGE_CALL_BACK_MAP = new ConcurrentHashMap<>();

    private static Map<String, Pair<Class<?>,Class<?>>> correlationDataForClass = new ConcurrentHashMap<>();

    private BeanFactory beanFactory;

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource(name = "doMqCallBackTaskExecutor")
    private AsyncTaskExecutor asyncTaskExecutor;

    public RabbitTemplateCallback(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
    }

    @PostConstruct
    public void init() {
        if (Objects.nonNull(rabbitTemplate)) {
            rabbitTemplate.setConfirmCallback(this);            //指定 ConfirmCallback
            rabbitTemplate.setReturnCallback(this);             //指定 ReturnCallback
        }
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        try {
            Pair<Class<?>, Class<?>> classClassPair = correlationDataForClass.get(correlationData.getId());
            Class<?> first = classClassPair.getFirst();
            Class<?> second = classClassPair.getSecond();
            if (Objects.nonNull(first)) {
                CallbackInterface callbackInterface = LAZY_MESSAGE_CALL_BACK_MAP.computeIfAbsent(first,k->(CallbackInterface)beanFactory.getBean(first));
                if (Objects.nonNull(callbackInterface)) {

                    CompletableFuture.runAsync(() -> {
                       // callbackInterface.confirm(correlationData, ack, cause,correlationData.getReturnedMessage().getBody());
                        callbackInterface.confirm(correlationData, ack, cause, JdkSerialization.deserialize(correlationData.getReturnedMessage().getBody(),second));
                    }, asyncTaskExecutor).exceptionally((e) -> {
                        log.error("do call back  method [ mq confirm method] error,error message is :" + e.getMessage());
                        return null;
                    });
                }
            }
        } catch (Exception e) {
            log.error("execute call back method error,error message is :" + e.getMessage());
        } finally {
            correlationDataForClass.remove(correlationData.getId());
        }

    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        String correlationId = headers.get("spring_returned_message_correlation").toString();
        Pair<Class<?>, Class<?>> classClassPair = correlationDataForClass.get(correlationId);
        Class<?> first = classClassPair.getFirst();
        if (Objects.nonNull(first)) {
            CallbackInterface callbackInterface = LAZY_MESSAGE_CALL_BACK_MAP.computeIfAbsent(first,k->(CallbackInterface)beanFactory.getBean(first));
            if (Objects.nonNull(callbackInterface)) {
                CompletableFuture.runAsync(() -> {
                    callbackInterface.returnedMessage(message, replyCode, replyText, exchange, routingKey);
                }, asyncTaskExecutor).exceptionally((e) -> {
                    log.error("do call back  method [ mq returnedMessage method] error,error message is :" + e.getMessage());
                    return null;
                });

            }
        }
    }

    public static void addCorrelationDataToTL(String id, Class<?> clas,Class<?> clas1) {
        correlationDataForClass.put(id, Pair.of(clas,clas1));
    }

}
